package com.neusoft.databus.agent.interactive.collect;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Map.Entry;
import java.util.Optional;

import com.google.common.collect.Multimap;
import com.neusoft.bizcore.web.utils.SpringUtil;
import com.neusoft.databus.agent.parser.modbus.ModBusParser;
import com.neusoft.databus.agent.pool.socket.SocketConnector;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.MetricBean;
import com.neusoft.databus.common.exception.DisconnectException;
import com.neusoft.databus.common.exception.RetryableException;
import com.neusoft.databus.common.parser.Parser;
import com.neusoft.databus.common.utils.DigitalUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * modbus collector through scoket
 *
 * @author sunchf
 * @date 2019年1月4日 下午5:31:26
 */
@Slf4j
public class SocketModbusCollector extends BasicCollector<Socket> {

    public SocketModbusCollector(final Parser parser) {
        super(parser);
        this.connector = SpringUtil.getBean(SocketConnector.class);
    }

    @Override
    public void sendCommand(final CommandBean command) {
        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("begin to communicate with socket server，{} metrics will be executed",
                    command.getResourceBean().stream().mapToInt(it -> it.getMetrics().size()).sum());
        }

        final ModBusParser mparser = (ModBusParser) this.parser;
        final Multimap<String, MetricBean> multimap = mparser.parseCollectRequest(command);
        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("after optimizing, it will communicate with socket server {} times",
                    multimap.keySet().size());
        }

        InputStream is = null;
        OutputStream os = null;
        try {
            is = this.conn.getInputStream();
            os = this.conn.getOutputStream();
        } catch (final IOException e) {
            try {
                // 关闭操作很重要
                this.conn.close();
            } catch (final IOException e1) {
            }
            throw new DisconnectException(e.getMessage(), e);
        }
        for (final String req : multimap.keySet()) {

            byte[] collectedData = null;
            try {
                collectedData = this.retrieve(req, is, os);
            } catch (final IOException e) {
                try {
                    // 关闭操作很重要
                    this.conn.close();
                } catch (final IOException e1) {
                }
                throw new DisconnectException(e.getMessage(), e);
            }

            // 某一条指令没返回来数据，继续执行其它指令
            if (null == collectedData) {
                SocketModbusCollector.log.warn("do not recevice any excepted response.");
                continue;
            }

            for (final MetricBean metricBean : multimap.get(req)) {
                metricBean.setOriginalValue(collectedData);
            }
        }

        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("end of communication with socket server，{} metric collected.",
                    command.getResourceBean().stream()
                            .mapToLong(it -> it.getMetrics().stream().filter(m -> m.isExecutedSuccess()).count())
                            .sum());
        }
    }

    public byte[] retrieve(final String req, final InputStream is, final OutputStream os) throws IOException {

        final long startTime = System.currentTimeMillis();
        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("send a request: {}", req);
        }

        final byte[] adu = DigitalUtils.hexStringToBytes(req);

        if (is.available() > 0) {
            final int length = is.available();
            is.skip(length);
            if (SocketModbusCollector.log.isWarnEnabled()) {
                SocketModbusCollector.log.warn("{} bytes are availabe in inputstream, skip them.", length);
            }
        }

        os.write(adu);
        os.flush();

        final int expectedResDataLength = ((ModBusParser) this.parser).expectedResDataLength(adu);
        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("expected response length is: {}", expectedResDataLength);
        }
        int m = -1;
        final byte[] tmp = new byte[1024];
        final ByteArrayOutputStream output = new ByteArrayOutputStream();
        // 有可能抛超时异常soTimeout设置
        while ((m = is.read(tmp)) > -1) {
            output.write(tmp, 0, m);
            output.flush();
            break;
        }

        final byte[] data = output.toByteArray();
        if (m == -1) {
            throw new IOException("end of file is detected");
        }

        // 一般是指令错误，或服务端返回特定格式的异常数据
        if (data.length < expectedResDataLength) {
            if (SocketModbusCollector.log.isWarnEnabled()) {
                SocketModbusCollector.log.warn("receive a response {}，but not long enough，elapsed {} ms.",
                        DigitalUtils.bytesToHexString(data),
                        System.currentTimeMillis() - startTime);
            }
            return null;
        }

        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("receive a response {}，elapsed {} ms.", DigitalUtils.bytesToHexString(data),
                    System.currentTimeMillis() - startTime);
        } else if (SocketModbusCollector.log.isInfoEnabled()) {
            SocketModbusCollector.log.info("receive a response，elapsed {} ms.", System.currentTimeMillis() - startTime);
        }
        return data;
    }

    @Override
    public void doControl(final CommandBean command) throws Exception {
        if (SocketModbusCollector.log.isDebugEnabled()) {
            SocketModbusCollector.log.debug("start to control task for {}.",
                    command.getResourceBean().get(0).getName());
            SocketModbusCollector.log.debug("borrow connection object from pool {} by {}", command.getConnType(),
                    command.getConnectionStr());
        }

        try {
            this.conn = this.connector.getConn(command.getConnection().asUsedConnection());
            this.tryControl(command);
        } catch (final RetryableException e) {
            throw e;
        } catch (final Exception e) {
            throw e;
        } finally {
            try {
                if (SocketModbusCollector.log.isDebugEnabled()) {
                    SocketModbusCollector.log.debug("return connection object to poll, {}", command.getConnection());
                }
                this.connector.returnConn(command.getConnection().asUsedConnection(), this.conn);
            } catch (final Exception e) {

            }
        }

    }

    private void tryControl(final CommandBean command) throws IOException {
        final ModBusParser mparser = (ModBusParser) this.parser;
        final Multimap<String, MetricBean> multimap = mparser.parseCollectRequest(command);

        final Optional<Entry<String, MetricBean>> entry = multimap.entries().stream().findFirst();

        final String req = entry.get().getKey();
        final byte[] response = this.retrieve(req, this.conn.getInputStream(), this.conn.getOutputStream());
        final String resHex = DigitalUtils.bytesToHexString(response);
        if (req.equalsIgnoreCase(resHex)) {
            entry.get().getValue().setOriginalValue(response);
        }
    }

}
